﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using SomeServices;
using AutoMapper;
using CServiceStack.ServiceClient;
using MySql.Data.MySqlClient;
using System.Diagnostics;
using System.Threading.Tasks;
using System.Threading;
using SomeContracts;
using System.IO;
using System.Net;
using Microsoft.IO;
using CServiceStack.Text;
using CServiceStack.Plugins.ProtoBuf;
using System.IO.MemoryMappedFiles;
using System.Collections.Concurrent;
using CServiceStack.Plugins.FastInfoset;
using CServiceStack.ServiceModel.Serialization;

namespace SomeClients
{
    class Program
    {
        static void Main(string[] args)
        {
            string flag = Console.ReadLine();
            //AutoMapperTest();
            //InvokeTingServiceAsync();
            //MySqlInsertThroughput();
            //InvokeTingServiceWithFallbackAsync();
            InvokeTingService(flag);
            Console.ReadKey();
            //client.
        }

        static void MySqlInsertThroughput()
        {
            Stopwatch sw = new Stopwatch();
            sw.Start();
            #region big string
            string bigString = @"On 15 June 2001, NuSphere sued MySQL AB, TcX DataKonsult AB and its original authors Michael (Monty) Widenius and David Axmark in U.S District Court in Boston for breach of contract, tortious interference with third party contracts and relationships and unfair competition.[66][67]
In 2002, MySQL AB sued Progress NuSphere for copyright and trademark infringement in United States district court. NuSphere had allegedly violated MySQLs copyright by linking MySQLs GPLed code with NuSphere Gemini table without being in compliance with the license.[68] After a preliminary hearing before Judge Patti Saris on 27 February 2002, the parties entered settlement talks and eventually settled.[69] After the hearing, FSF commented that Judge Saris made clear that she sees the GNU GPL to be an enforceable and binding license.[70]
In October 2005, Oracle Corporation acquired Innobase OY, the Finnish company that developed the third-party InnoDB storage engine that allows MySQL to provide such functionality as transactions and foreign keys. After the acquisition, an Oracle press release mentioned that the contracts that make the companys software available to MySQL AB would be due for renewal (and presumably renegotiation) some time in 2006.[71] During the MySQL Users Conference in April 2006, MySQL issued a press release that confirmed that MySQL and Innobase OY agreed to a multi-year extension of their licensing agreement.[72]
In February 2006, Oracle Corporation acquired Sleepycat Software,[73] makers of the Berkeley DB, a database engine providing the basis for another MySQL storage engine. This had little effect, as Berkeley DB was not widely used, and was dropped (due to lack of use) in MySQL 5.1.12, a pre-GA release of MySQL 5.1 released in October 2006.[74]
In January 2008, Sun Microsystems bought MySQL for $1 billion.[75]
In April 2009, Oracle Corporation entered into an agreement to purchase Sun Microsystems,[76] then owners of MySQL copyright and trademark. Suns board of directors unanimously approved the deal, it was also approved by Suns shareholders, and by the U.S. government on 20 August 2009.[77] On 14 December 2009, Oracle pledged to continue to enhance MySQL[78] as it had done for the previous four years.
A movement against Oracles acquisition of MySQL, to Save MySQL[79] from Oracle was started by one of the MySQL founders, Monty Widenius. The petition of 50,000+ developers and users called upon the European Commission to block approval of the acquisition. At the same time, several Free Software opinion leaders (including Eben Moglen, Pamela Jones of Groklaw, Jan Wildeboer and Carlo Piana, who also acted as co-counsel in the merger regulation procedure) advocated for the unconditional approval of the merger.[citation needed] As part of the negotiations with the European Commission, Oracle committed that MySQL server will continue until at least 2015 to use the dual-licensing strategy long used by MySQL AB, with proprietary and GPL versions available. The antitrust of the EU had been pressuring it to divest MySQL as a condition for approval of the merger. But, as revealed by WikiLeaks, the US Department of Justice and Antitrust, at the request of Oracle, pressured the EU to unconditionally approve the merger.[80] The European Commission eventually unconditionally approved Oracles acquisition of MySQL on 21 January 2010.[81]
In January 2009, prior to Oracles acquisition of MySQL, Monty Widenius started a GPL-only fork, MariaDB. MariaDB is based on the same code base as MySQL server 5.1 and strives to maintain compatibility with Oracle-provided versions.On 15 June 2001, NuSphere sued MySQL AB, TcX DataKonsult AB and its original authors Michael (Monty) Widenius and David Axmark in U.S District Court in Boston for breach of contract, tortious interference with third party contracts and relationships and unfair competition.[66][67]
In 2002, MySQL AB sued Progress NuSphere for copyright and trademark infringement in United States district court. NuSphere had allegedly violated MySQLs copyright by linking MySQLs GPLed code with NuSphere Gemini table without being in compliance with the license.[68] After a preliminary hearing before Judge Patti Saris on 27 February 2002, the parties entered settlement talks and eventually settled.[69] After the hearing, FSF commented that Judge Saris made clear that she sees the GNU GPL to be an enforceable and binding license.[70]
In October 2005, Oracle Corporation acquired Innobase OY, the Finnish company that developed the third-party InnoDB storage engine that allows MySQL to provide such functionality as transactions and foreign keys. After the acquisition, an Oracle press release mentioned that the contracts that make the companys software available to MySQL AB would be due for renewal (and presumably renegotiation) some time in 2006.[71] During the MySQL Users Conference in April 2006, MySQL issued a press release that confirmed that MySQL and Innobase OY agreed to a multi-year extension of their licensing agreement.[72]
In February 2006, Oracle Corporation acquired Sleepycat Software,[73] makers of the Berkeley DB, a database engine providing the basis for another MySQL storage engine. This had little effect, as Berkeley DB was not widely used, and was dropped (due to lack of use) in MySQL 5.1.12, a pre-GA release of MySQL 5.1 released in October 2006.[74]
In January 2008, Sun Microsystems bought MySQL for $1 billion.[75]
In April 2009, Oracle Corporation entered into an agreement to purchase Sun Microsystems,[76] then owners of MySQL copyright and trademark. Suns board of directors unanimously approved the deal, it was also approved by Suns shareholders, and by the U.S. government on 20 August 2009.[77] On 14 December 2009, Oracle pledged to continue to enhance MySQL[78] as it had done for the previous four years.
A movement against Oracles acquisition of MySQL, to Save MySQL[79] from Oracle was started by one of the MySQL founders, Monty Widenius. The petition of 50,000+ developers and users called upon the European Commission to block approval of the acquisition. At the same time, several Free Software opinion leaders (including Eben Moglen, Pamela Jones of Groklaw, Jan Wildeboer and Carlo Piana, who also acted as co-counsel in the merger regulation procedure) advocated for the unconditional approval of the merger.[citation needed] As part of the negotiations with the European Commission, Oracle committed that MySQL server will continue until at least 2015 to use the dual-licensing strategy long used by MySQL AB, with proprietary and GPL versions available. The antitrust of the EU had been pressuring it to divest MySQL as a condition for approval of the merger. But, as revealed by WikiLeaks, the US Department of Justice and Antitrust, at the request of Oracle, pressured the EU to unconditionally approve the merger.[80] The European Commission eventually unconditionally approved Oracles acquisition of MySQL on 21 January 2010.[81]
In January 2009, prior to Oracles acquisition of MySQL, Monty Widenius started a GPL-only fork, MariaDB. MariaDB is based on the same code base as MySQL server 5.1 and strives to maintain compatibility with Oracle-provided versions.On 15 June 2001, NuSphere sued MySQL AB, TcX DataKonsult AB and its original authors Michael (Monty) Widenius and David Axmark in U.S District Court in Boston for breach of contract, tortious interference with third party contracts and relationships and unfair competition.[66][67]
In 2002, MySQL AB sued Progress NuSphere for copyright and trademark infringement in United States district court. NuSphere had allegedly violated MySQLs copyright by linking MySQLs GPLed code with NuSphere Gemini table without being in compliance with the license.[68] After a preliminary hearing before Judge Patti Saris on 27 February 2002, the parties entered settlement talks and eventually settled.[69] After the hearing, FSF commented that Judge Saris made clear that she sees the GNU GPL to be an enforceable and binding license.[70]
In October 2005, Oracle Corporation acquired Innobase OY, the Finnish company that developed the third-party InnoDB storage engine that allows MySQL to provide such functionality as transactions and foreign keys. After the acquisition, an Oracle press release mentioned that the contracts that make the companys software available to MySQL AB would be due for renewal (and presumably renegotiation) some time in 2006.[71] During the MySQL Users Conference in April 2006, MySQL issued a press release that confirmed that MySQL and Innobase OY agreed to a multi-year extension of their licensing agreement.[72]
In February 2006, Oracle Corporation acquired Sleepycat Software,[73] makers of the Berkeley DB, a database engine providing the basis for another MySQL storage engine. This had little effect, as Berkeley DB was not widely used, and was dropped (due to lack of use) in MySQL 5.1.12, a pre-GA release of MySQL 5.1 released in October 2006.[74]
In January 2008, Sun Microsystems bought MySQL for $1 billion.[75]
In April 2009, Oracle Corporation entered into an agreement to purchase Sun Microsystems,[76] then owners of MySQL copyright and trademark. Suns board of directors unanimously approved the deal, it was also approved by Suns shareholders, and by the U.S. government on 20 August 2009.[77] On 14 December 2009, Oracle pledged to continue to enhance MySQL[78] as it had done for the previous four years.";
            #endregion
            var result = Parallel.For(0, 100, (i) =>
            //for (int i = 0; i < 1; i++)
            {
                using (MySqlConnection connection = new MySqlConnection("Server=localhost;Database=test;Uid=root;Pwd=P@ssw0rd;"))
                {
                    connection.Open();
                    //Parallel.For(0,10,(i)=>
                    ////for (int i = 0; i < 1; i++)
                    //{
                    MySqlCommand cmd = connection.CreateCommand();
                    StringBuilder cmdStr = new StringBuilder();
                    cmdStr.Append("insert into inserttable(field1,field2) ");
                    cmdStr.Append("values('");
                    cmdStr.Append(bigString);
                    cmdStr.Append("','");
                    cmdStr.Append(bigString);
                    cmdStr.Append("')");
                    for (int j = 0; j < 50; j++)
                    {
                        cmdStr.Append(",('");
                        cmdStr.Append(bigString);
                        cmdStr.Append("','");
                        cmdStr.Append(bigString);
                        cmdStr.Append("')");
                    }
                    cmdStr.Append(";");
                    cmd.CommandText = cmdStr.ToString();
                    cmd.ExecuteNonQuery();
                    //});                
                }
            });
            while (!result.IsCompleted)
            {
                Thread.Sleep(100);
            }
            sw.Stop();
            Console.WriteLine("write 100 M takes(ms):" + sw.ElapsedMilliseconds);
        }

        private static void AutoMapperTest()
        {
            Order order = new Order() { OrderID = 1, Amount = 10, LastChanged = DateTime.Now };
            OrderExt orderEx = new OrderExt() { OrderID = 1, LastChanged = DateTime.Now, Status = 2 };

            OrderDTO dto = Mapper.DynamicMap<OrderDTO>(order);
            Mapper.DynamicMap<OrderExt, OrderDTO>(orderEx, dto);
            Console.WriteLine(string.Format("orderID:{0},lastchanged:{1},amount:{2},status:{3}", dto.OrderID, dto.LastChanged, dto.Amount, dto.Status));
        }

        private static void InvokeTingService(string flag)
        {
            //TingServiceClient.RegisterCallFormat(new SomeCallFormat());
            //TingServiceClient.RegisterCallFormat(new SomeProtoBufClientCallFormat());
            TingServiceClient client = TingServiceClient.GetInstance("http://localhost:2013/");
            //TingServiceClient client = new TingServiceClient();

            IClientCallFormat callFormat;
            if (flag == "1")
            {
                callFormat = new JsonClientCallFormat();//new ProtoBufClientCallFormat();//new FastInfosetClientCallFormat();//
            }
            else
            {
                callFormat = new SomeCallFormat();//new SomeProtoBufClientCallFormat();
                //callFormat = new MMFProtoBufClientCallFormat();
                //callFormat = new MMFCallFormat();
            }
            TingServiceClient.RegisterCallFormat(callFormat);
            client.Format = callFormat.Format;
            Console.WriteLine(client.Format);
            //TingServiceClient.HttpWebRequestFilter = new Action<System.Net.HttpWebRequest>(request => {
            //    request.AddRange(0, 2);
            //});
            //TingServiceClient.HttpWebResponseFilter = new Action<System.Net.HttpWebResponse>((resp) =>
            //{
            //    using (Stream responseStream = resp.GetResponseStream())
            //    {
            //        byte[] buffer = new byte[1024];
            //        int bytesRead = 0;
            //        while ((bytesRead = responseStream.Read(buffer, 0, buffer.Length)) > 0)
            //        {
            //            Console.WriteLine(bytesRead);
            //        }
            //    }
            //});
            //var response = client.Ring(new RingRequestType() { RingName = "ting" });
            //Console.WriteLine(response.result.Count);
            //response = client.Ring(new RingRequestType() { RingName = "ting" });
            //Console.WriteLine(response.result.Count);
            Stopwatch sw = new Stopwatch();
            sw.Start();
            Parallel.For(0, 1000, new ParallelOptions { MaxDegreeOfParallelism = 8 }, (i) =>
            {
                //RingResponseType response = null;
                RingRequestType request = new RingRequestType() { RingName = "ting" };
                //for (int i = 0; i < 1000; i++)
                //{
                //response = 
                client.Ring(request);
                Console.WriteLine("ok");//response.result.Times);

                //}
            });
            sw.Stop();
            Console.WriteLine("takes:" + sw.Elapsed.TotalSeconds + " s");
            //Thread.Sleep(500);
        }



        private static void InvokeTingServiceAsync()
        {
            TingServiceClient client = TingServiceClient.GetInstance("http://localhost:2013/");
            //TingServiceClient client = new TingServiceClient();
            client.Format = "json";
            for (int i = 0; i < 1000; i++)
            {
                //var task = client.CreateAsyncTaskOfRing(new RingRequestType() { ringName = "ting" });
                //task.ContinueWith((t) =>
                //{
                //    Console.WriteLine(t.Result.result.Count);
                //});
                //task.Start();
                client.StartIOCPTask<RingResponseType>("Ring", new RingRequestType() { RingName = "ting" }).ContinueWith((t) =>
                {
                    if (t.Exception == null)
                    {
                        Console.WriteLine(t.Result.result.Times);
                    }
                    else
                    {
                        foreach (var ex in t.Exception.InnerExceptions)
                        {
                            var soaEx = ex as WebProtocolException;
                            if (soaEx != null)
                            {
                                Console.WriteLine(soaEx.StatusCode + ":" + soaEx.StatusDescription);
                            }
                            Console.WriteLine(ex);
                        }
                    }
                });
            }
        }

        private static void InvokeTingServiceWithFallbackAsync()
        {
            int fallbackCounter = 0;
            for (int j = 0; j < 10; j++)
            {
                Console.WriteLine("batch:" + j);
                TingServiceClient client = TingServiceClient.GetInstance("http://localhost:2013/");

                client.Format = "json";
                int total = 100;
                int failedCounter = total;
                Task[] tasks = new Task[total];
                for (int i = 0; i < total; i++)
                {

                    var task = client.StartIOCPTask<RingResponseType>("Ring", new RingRequestType() { RingName = "ting" });
                    tasks[i] = task;
                    task.ContinueWith((t) =>
                    {
                        if (t.Exception == null)
                        {
                            Interlocked.Decrement(ref failedCounter);
                            Console.WriteLine(t.Result.ToString());
                        }
                        else
                        {
                            Console.WriteLine(t.Exception);
                        }
                    });
                }
                Task.WaitAll(tasks, 1000);

                if (failedCounter >= 50)
                {
                    //log & send mail & break;     
                    Console.WriteLine("too many fails");
                    return;
                }
                if (failedCounter >= 20)
                {
                    fallbackCounter++;
                    if (fallbackCounter >= 3)
                    {
                        //log & send mail & break;
                        Console.WriteLine("too many fallbacks");
                        return;
                    }
                    Thread.Sleep(10000 * fallbackCounter);
                }
            }

        }
    }

    public class SomeCallFormat : IClientCallFormat
    {
        ClientStreamDeserializerDelegate _deserializerDelegate;
        ClientStreamSerializerDelegate _serializerDelegate;
        static readonly RecyclableMemoryStreamManager _rmsMgr = new RecyclableMemoryStreamManager(200 * 1024, 1024 * 1024, 128 * 1024 * 1024);
        static ConcurrentQueue<byte[]> bufferQueue;
        static ConcurrentQueue<char[]> charBufferQueue;
        static SomeCallFormat()
        {
            bufferQueue = new ConcurrentQueue<byte[]>();
            charBufferQueue = new ConcurrentQueue<char[]>();
            for (int i = 0; i < 10; i++)
            {
                Byte[] buffer = new byte[_rmsMgr.BlockSize];
                bufferQueue.Enqueue(buffer);
                char[] charBuffer = new char[128];
                charBufferQueue.Enqueue(charBuffer);
            }
        }

        public SomeCallFormat()
        {
            JsonClientCallFormat format = new JsonClientCallFormat();
            _deserializerDelegate = DeserializeFromStream;
            //    new ClientStreamDeserializerDelegate((t, stream) =>
            //{
            //    MemoryStream innerStream = _rmsMgr.GetStream();
            //    byte[] buffer;
            //    while (!bufferQueue.TryDequeue(out buffer))
            //    {
            //        Thread.Sleep(200);
            //    }//new byte[_rmsMgr.BlockSize];
            //    //byte[] buffer = new byte[512];
            //    //int bytesRead;
            //    //while ((bytesRead = stream.Read(buffer, 0, buffer.Length)) > 0)
            //    //{
            //    //    innerStream.Write(buffer, 0, bytesRead);
            //    //}
            //    stream.CopyTo(innerStream, buffer);
            //    bufferQueue.Enqueue(buffer);
            //    innerStream.Seek(0, SeekOrigin.Begin);
            //    //String result = Encoding.UTF8.GetString(innerStream.ToArray());
            //    //return WrappedXmlSerializer.DeserializeFromString(result, t);
            //    return format.StreamDeserializer(t, innerStream);
            //    //return WrappedJsonDeserializer.Instance.DeserializeFromStream(t, innerStream);
            //    //return format.StreamDeserializer(t, innerStream);

            //});
            _serializerDelegate = format.StreamSerializer;
            //new ClientStreamSerializerDelegate((o, stream) =>
            // {
            //     //MemoryStream innerStream = rmsMgr.GetStream();
            //     //byte[] buffer = new byte[512];
            //     //while (stream.Read(buffer, 0, buffer.Length) > 0)
            //     //{
            //     //    innerStream.Write(buffer, 0, buffer.Length);
            //     //}
            //     format.StreamSerializer(o, stream);
            // });
        }
        public string ContentType
        {
            get { return "application/some"; }
        }

        public string Format
        {
            get { return "some"; }
        }

        public ClientStreamDeserializerDelegate StreamDeserializer
        {
            get { return _deserializerDelegate; }
        }

        public ClientStreamSerializerDelegate StreamSerializer
        {
            get { return _serializerDelegate; }
        }

        public static object DeserializeFromStream(Type type, Stream stream)
        {
            using (var reader = new StreamReader(stream, new UTF8Encoding(false)))
            {
                char[] charBuffer;
                while (!charBufferQueue.TryDequeue(out charBuffer))
                {
                    Thread.Sleep(200);
                }
                StringBuilder content = new StringBuilder();
                int count = 0;count = reader.ReadBlock(charBuffer, 0, charBuffer.Length);
                while ( count> 0)
                {
                    content.Append(charBuffer, 0, count);
                    count = reader.ReadBlock(charBuffer, 0, charBuffer.Length);
                }
                charBufferQueue.Enqueue(charBuffer);
                return JsonSerializer.DeserializeFromString( content.ToString(),type);
                //return DeserializeFromString(reader.ReadToEnd(), type);
            }
        }
    }

    public class MMFCallFormat : IClientCallFormat
    {
        ClientStreamDeserializerDelegate _deserializerDelegate;
        ClientStreamSerializerDelegate _serializerDelegate;


        public MMFCallFormat()
        {
            XmlClientCallFormat format = new XmlClientCallFormat();
            _deserializerDelegate = new ClientStreamDeserializerDelegate((t, stream) =>
            {
                using (MemoryMappedFile mmf = MemoryMappedFile.CreateNew(this.GetHashCode().ToString(), 80 * 1024 * 1024))
                {
                    using (MemoryMappedViewStream innerStream = mmf.CreateViewStream())
                    {
                        byte[] buffer = new byte[512];
                        int bytesRead;
                        while ((bytesRead = stream.Read(buffer, 0, buffer.Length)) > 0)
                        {
                            innerStream.Write(buffer, 0, bytesRead);
                        }
                        innerStream.Seek(0, SeekOrigin.Begin);

                        return WrappedXmlSerializer.DeserializeFromStream(t, innerStream);
                    }
                }
                //String result = Encoding.UTF8.GetString(innerStream.ToArray());
                //return WrappedXmlSerializer.DeserializeFromString(result, t);
                //return format.StreamDeserializer(t, innerStream);

            });
            _serializerDelegate = new ClientStreamSerializerDelegate((o, stream) =>
            {
                //MemoryStream innerStream = rmsMgr.GetStream();
                //byte[] buffer = new byte[512];
                //while (stream.Read(buffer, 0, buffer.Length) > 0)
                //{
                //    innerStream.Write(buffer, 0, buffer.Length);
                //}
                format.StreamSerializer(o, stream);
            });
        }
        public string ContentType
        {
            get { return "application/mmf"; }
        }

        public string Format
        {
            get { return "mmf"; }
        }

        public ClientStreamDeserializerDelegate StreamDeserializer
        {
            get { return _deserializerDelegate; }
        }

        public ClientStreamSerializerDelegate StreamSerializer
        {
            get { return _serializerDelegate; }
        }
    }
    public class SomeProtoBufClientCallFormat : IClientCallFormat
    {
        ProtoBufClientCallFormat _format = new ProtoBufClientCallFormat();
        ClientStreamDeserializerDelegate _streamDeserializer;
        static readonly int blockSize = 80 * 1024;
        static readonly int largeBufferMultiple = 1024 * 1024;
        static readonly int maxBufferSize = 100 * largeBufferMultiple;
        static readonly RecyclableMemoryStreamManager _rmsMgr;
        static ConcurrentQueue<byte[]> bufferQueue;

        static SomeProtoBufClientCallFormat()
        {
            _rmsMgr = new RecyclableMemoryStreamManager(blockSize, largeBufferMultiple, maxBufferSize);
            //_rmsMgr.AggressiveBufferReturn = true;
            bufferQueue = new ConcurrentQueue<byte[]>();
            for (int i = 0; i < 10; i++)
            {
                Byte[] buffer = new byte[_rmsMgr.LargeBufferMultiple];
                bufferQueue.Enqueue(buffer);
            }
        }

        public SomeProtoBufClientCallFormat()
        {
            _streamDeserializer = new ClientStreamDeserializerDelegate((type, stream) =>
            {
                using (MemoryStream innerStream = _rmsMgr.GetStream("SomeProtoBufClientCallFormat", _rmsMgr.LargeBufferMultiple, true))
                {
                    byte[] buffer;
                    while (!bufferQueue.TryDequeue(out buffer))
                    {
                        Thread.Sleep(200);
                    }//new byte[_rmsMgr.BlockSize];
                    stream.CopyTo(innerStream, buffer);// _rmsMgr.LargeBufferMultiple);
                    stream.Dispose();
                    //BinaryReader reader = new BinaryReader(stream);
                    //reader.read

                    //int bytesRead;
                    //int totalBytes = 0;
                    //while ((bytesRead = stream.Read(buffer, 0, buffer.Length)) > 0)
                    //{
                    //    totalBytes += bytesRead;
                    //    innerStream.Write(buffer, 0, bytesRead);
                    //}
                    bufferQueue.Enqueue(buffer);
                    innerStream.Seek(0, SeekOrigin.Begin);
                    //Console.WriteLine(totalBytes / (1024 * 1024)+"MB");
                    return _format.StreamDeserializer(type, innerStream);
                }
            });
        }
        public string ContentType
        {
            get { return "application/someprotobuf"; }
        }

        public string Format
        {
            get { return "someprotobuf"; }
        }

        public ClientStreamDeserializerDelegate StreamDeserializer
        {
            get { return _streamDeserializer; }
        }

        public ClientStreamSerializerDelegate StreamSerializer
        {
            get { return _format.StreamSerializer; }
        }
    }

    public class MMFProtoBufClientCallFormat : IClientCallFormat
    {
        ProtoBufClientCallFormat _format = new ProtoBufClientCallFormat();
        ClientStreamDeserializerDelegate _streamDeserializer;


        public MMFProtoBufClientCallFormat()
        {
            _streamDeserializer = new ClientStreamDeserializerDelegate((type, stream) =>
            {

                using (MemoryMappedFile mmf = MemoryMappedFile.CreateNew(this.GetHashCode().ToString(), 20 * 1024 * 1024))
                {
                    using (MemoryMappedViewStream innerStream = mmf.CreateViewStream())
                    {
                        byte[] buffer = new byte[512];
                        int bytesRead;
                        while ((bytesRead = stream.Read(buffer, 0, buffer.Length)) > 0)
                        {
                            innerStream.Write(buffer, 0, bytesRead);
                        }
                        innerStream.Seek(0, SeekOrigin.Begin);

                        return _format.StreamDeserializer(type, innerStream);
                    }
                }

            });
        }
        public string ContentType
        {
            get { return "application/mmfprotobuf"; }
        }

        public string Format
        {
            get { return "mmfprotobuf"; }
        }

        public ClientStreamDeserializerDelegate StreamDeserializer
        {
            get { return _streamDeserializer; }
        }

        public ClientStreamSerializerDelegate StreamSerializer
        {
            get { return _format.StreamSerializer; }
        }
    }

}
